package t20240424;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class Test2 {


    private static final Logger LOG = LoggerFactory.getLogger(Test2.class);
    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setString(RestOptions.BIND_PORT,"8080");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


        String source = "CREATE TABLE source_table (\n" +
                "    id INT,\n" +
                "    k1 STRING,\n" +
                "    v1 STRING\n" +
                ") WITH (\n" +
                "    'connector' = 'datagen',\n" +
                "    'rows-per-second' = '5'\n" +
                ")";

        String sink = "CREATE TABLE pprint (\n" +
                "    id INT,\n" +
                "    kv STRING\n" +
                ") WITH (\n" +
                "    'connector' = 'print'\n" +
                ")";


        String pro = "INSERT INTO pprint\n" +
                "SELECT id, k1 FROM source_table";



        tableEnv.executeSql(source);
        tableEnv.executeSql(sink);
        tableEnv.executeSql(pro);



    }
}
